package com.zetyun.tiger.connector.amt;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;

import java.math.BigDecimal;

import static com.zetyun.tiger.connector.op.v2.RedisOpV2Action.build;

//@ActionMeta(
//        id = "tiger.count-amt",
//        tags = {"redis"},
//        category = "map",
//        name = "CountAmtAction",
//        description = "计算今日客户经理与机构公共揽存金额",
//        parameterClass = CountAmtParameter.class,
//        status = ActionMeta.Status.PUBLIC
//)
@Slf4j
public class CountAmtAction extends MapAction {

    private CountAmtParameter param;

    private SimpleRedisContainer rc;

    private String caKey;

    private String oaKey;

    private String dateField;

    private String orgField;

    private String conntrField;

    private String amtField;

    private String rtConntrAmtField;

    private String rtOrgPdAmtField;

    private String rtOrgAmtField;

    private String pubDepFlag;

    private String pemsOrgKey;

    private String pemsOrgField;

    private final Integer ttl = 60 * 60 * 24;

    @Override
    public void init(OperatorContext context) {
        super.init(context);


        param = context.getParameters();
        caKey = param.getCaKey();
        oaKey = param.getOaKey();
        dateField = param.getDateField();
        orgField = param.getOrgField();
        conntrField = param.getConntrField();
        amtField = param.getAmtField();
        rtConntrAmtField = param.getRtConntrAmtField();
        rtOrgPdAmtField = param.getRtOrgPdAmtField();
        pubDepFlag = param.getPubDepFlag();
        pemsOrgKey = param.getPemsOrgKey();
        pemsOrgField = param.getOrgField();
        rtOrgAmtField = param.getRtOrgAmtField();
        FlinkJedisSentinelConfig conf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(param.getMasterName())
                        .setSentinels(param.getSentinels())
                        .setPassword(param.getPassword())
                        .setDatabase(param.getDatabase())
                        .build();
        rc = build(conf);
        try {
            rc.open();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public RtEvent apply(RtEvent in) throws Exception {

        String dateValue = in.getField(dateField, String.class);
        String pemsOrgValue = in.getField(pemsOrgField, String.class);
        String orgValue = in.getField(orgField, String.class);
        String conntrValue = in.getField(conntrField, String.class);
        String amtStrValue = in.getField(amtField, String.class);

        BigDecimal amtValue = new BigDecimal(amtStrValue);

        if (!conntrValue.isEmpty()) {
            // 表明为此数据归属于客户经理
            String caDayCacheKey = caKey.concat(dateValue);
            String pemsOrgDayCacheKey = pemsOrgKey.concat(dateValue);
            BigDecimal currentAmt = hincrBy(caDayCacheKey, conntrValue, amtValue);

            // 获取该机构下所有客户经理的总金额
            BigDecimal currentPemsOrgAmt = hincrBy(pemsOrgDayCacheKey, pemsOrgValue, amtValue);
            // 获取该机构的公共揽存金额
            String oaDayCacheKey = oaKey.concat(dateValue);
            BigDecimal currentCaAmt = hgetPemsOrgAmt(oaDayCacheKey, pemsOrgValue);

            // 两个金额相加即得 该机构下的所有金额
            BigDecimal rtOrgAmt = currentPemsOrgAmt.add(currentCaAmt);

            in.setValue(rtConntrAmtField, currentAmt);
            in.setValue(rtOrgPdAmtField, "");
            in.setValue(rtOrgAmtField, rtOrgAmt);
            in.setValue(pubDepFlag, "N");
        } else {
            // 表明为此数据归属于公共揽存
            String oaDayCacheKey = oaKey.concat(dateValue);
            BigDecimal currentAmt = hincrBy(oaDayCacheKey, orgValue, amtValue);

            // 获取机构下客户经理总金额
            String pemsOrgDayCacheKey = pemsOrgKey.concat(dateValue);
            BigDecimal allConntrAmt = hgetPemsOrgAmt(pemsOrgDayCacheKey, orgValue);

            // 获取机构总金额
            BigDecimal rtOrgAmt = allConntrAmt.add(currentAmt);

            in.setValue(rtConntrAmtField, "");
            in.setValue(rtOrgPdAmtField, currentAmt);
            in.setValue(rtOrgAmtField, rtOrgAmt);
            in.setValue(pubDepFlag, "Y");
        }

        return in;
    }

    private BigDecimal hincrBy(String key, String hashField, BigDecimal value) {
        String lastValue = rc.hget(key, hashField);
        if (lastValue == null) {
            rc.hset(key, hashField, value.toString(), ttl);
            return value;
        } else {
            BigDecimal currentVal = value.add(new BigDecimal(lastValue));
            rc.hset(key, hashField, currentVal.toString(), ttl);
            return currentVal;
        }

    }

    private BigDecimal hgetPemsOrgAmt(String key, String hashField) {
        String lastValue = rc.hget(key, hashField);
        if (lastValue == null) {
            return BigDecimal.ZERO;
        } else {
            return new BigDecimal(lastValue);
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        rc.close();
    }
}
